package com.yifeng.repo.base.utils.async;

import com.yifeng.repo.base.utils.common.BaseUtil;
import org.apache.skywalking.apm.toolkit.trace.CallableWrapper;
import org.apache.skywalking.apm.toolkit.trace.RunnableWrapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 线程执行器
 * Created by daibing on 2020/3/8.
 */
public class FireExecutor {
    private static final Logger LOGGER = LoggerFactory.getLogger(FireExecutor.class);
    private static final Integer STAT_THRESHOLD = 100;

    private final AtomicInteger count = new AtomicInteger(0);
    private final ThreadPoolExecutor executor;
    private final Integer corePoolSize;
    private final Integer maximumPoolSize;
    private final Integer queueSize;
    private final String threadNamePrefix;

    public FireExecutor(Integer corePoolSize, Integer maximumPoolSize, Integer queueSize, String threadNamePrefix) {
        this.corePoolSize = corePoolSize == null ? 1 : corePoolSize;
        this.maximumPoolSize = maximumPoolSize == null ? 2 : maximumPoolSize;
        this.queueSize = queueSize == null ? 10 : queueSize;
        this.threadNamePrefix = BaseUtil.isBlank(threadNamePrefix) ? "fire-exec" : threadNamePrefix;
        this.executor = this.init();
    }

    private ThreadPoolExecutor init() {
        return new ThreadPoolExecutor(
                corePoolSize,
                maximumPoolSize,
                60L,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(queueSize),
                new ThreadFactory() {
                    final Random random = new Random();

                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r);
                        thread.setName(threadNamePrefix + this.random.nextInt());
                        return thread;
                    }
                },
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    public void shutdown() {
        this.printPoolStatus(count.get());
        this.executor.shutdown();
    }

    public <T> void executeTask(Runnable task) {
        this.countNumber();
        if (BaseUtil.EXIST_SKYWALKING_WRAPPER) {
            this.executor.execute(RunnableWrapper.of(task));
        } else {
            this.executor.execute(task);
        }
    }

    public <T> void batchExecuteTask(Collection<? extends Runnable> tasks) {
        for (Runnable task : tasks) {
            executeTask(task);
        }
    }

    public <T> Future<T> submitTask(Callable<T> task) {
        this.countNumber();
        if (BaseUtil.EXIST_SKYWALKING_WRAPPER) {
            return this.executor.submit(CallableWrapper.of(task));
        } else {
            return this.executor.submit(task);
        }
    }

    public <T> T runTask(Callable<T> task) {
        this.countNumber();
        Future<T> future = BaseUtil.EXIST_SKYWALKING_WRAPPER ? this.executor.submit(CallableWrapper.of(task)) : this.executor.submit(task);
        try {
            return future.get();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public <T> List<T> batchRunTask(Collection<? extends Callable<T>> tasks) {
        List<Future<T>> futureList = this.invokeAll(tasks);
        List<T> result = new ArrayList<>(futureList.size());
        try {
            for (Future<T> future : futureList) {
                result.add(future.get());
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return result;
    }

    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) {
        this.countNumber(tasks.size());
        try {
            if (BaseUtil.EXIST_SKYWALKING_WRAPPER) {
                List<CallableWrapper<T>> wrappers = new ArrayList<>();
                for (Callable<T> task : tasks) {
                    wrappers.add(CallableWrapper.of(task));
                }
                return this.executor.invokeAll(wrappers);
            } else {
                return this.executor.invokeAll(tasks);
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    public void printPoolStatus() {
        this.printPoolStatus(count.get());
    }

    private void printPoolStatus(int count) {
        LOGGER.info("ThreadName={}, Count={}, PoolSize={}, LargestPoolSize={}, TaskCount={}, CompletedTaskCount={}, ActiveCount={}, workQueueSize={}",
                threadNamePrefix,
                count,
                executor.getPoolSize(),
                executor.getLargestPoolSize(),
                executor.getTaskCount(),
                executor.getCompletedTaskCount(),
                executor.getActiveCount(),
                executor.getQueue().size()
        );
        // temp test ...
        // System.out.println("countNumber2: " + count);
    }

    private void countNumber() {
        if (STAT_THRESHOLD.compareTo(count.incrementAndGet()) <= 0) {
            this.printPoolStatus(count.getAndSet(0));
        }
        // temp test ...
        // System.out.println("countNumber1: " + count.get());
    }

    private void countNumber(int delta) {
        if (STAT_THRESHOLD.compareTo(count.addAndGet(delta)) <= 0) {
            this.printPoolStatus(count.getAndSet(0));
        }
    }

    public static void main(String[] args) throws Exception {
        System.out.println("main start to do...");
        final SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss.SSS");
        final AtomicInteger number = new AtomicInteger(0);
        FireExecutor executor = new FireExecutor(1, 1, 128, "my-exec");

        System.out.println("start to run...");
        // callable接口实现类中的run方法允许异常向上抛出，可以在内部处理，try catch
        for (int i = 0; i < 1000; i++) {
            executor.submitTask(() -> {
                System.out.printf("Callable thread=%s, time=%s, number=%s%n", Thread.currentThread().getName(), sdf.format(new Date()), number.incrementAndGet());
                throw new IllegalArgumentException("尝试错误1！");
            });
        }
        // runnable接口实现类中run方法的异常必须在内部处理，不能抛出
        for (int i = 0; i < 1000; i++) {
            executor.executeTask(() -> {
                System.out.printf("Runnable thread=%s, time=%s, number=%s%n", Thread.currentThread().getName(), sdf.format(new Date()), number.incrementAndGet());
                throw new IllegalArgumentException("尝试错误2！");
            });
        }
        System.in.read();
    }
}
